package com.silky.officeline.thread;


import com.alibaba.fastjson.JSON;
import com.silky.common.bean.DataSourceConfig;
import com.silky.common.bean.MultiLevelJoinConfig;
import com.silky.common.bean.NodeConfig;
import com.silky.common.service.NodeService;
import com.silky.common.utils.PageList;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Join结果计算线程
 */
@Slf4j
public class JoinResultCalculationThread extends AbstractThread {

    private final DataSourceConfig dataSourceConfig;
    private final MultiLevelJoinConfig multiLevelJoinConfig;
    private final PageList currentPageList;
    private final LinkedBlockingQueue<List<Map<String, Object>>> queue;
    private final int currentLevel;
    private final List<Map<String, Object>> parentNodeDatas;
    private final String parentNodeDataAlias;
    private NodeService nodeService;
    private Connection connection=null;

    public JoinResultCalculationThread(DataSourceConfig dataSourceConfig, MultiLevelJoinConfig multiLevelJoinConfig, PageList currentPageList, LinkedBlockingQueue<List<Map<String, Object>>> queue, int currentLevel, List<Map<String, Object>> parentNodeDatas, String parentNodeDataAlias) {
        this.dataSourceConfig = dataSourceConfig;
        this.multiLevelJoinConfig = multiLevelJoinConfig;
        this.currentPageList = currentPageList;
        this.queue = queue;
        this.currentLevel = currentLevel;
        this.parentNodeDatas = parentNodeDatas;
        nodeService = new NodeService(dataSourceConfig);
        this.parentNodeDataAlias = parentNodeDataAlias;
    }

    @Override
    public void run(){
        System.out.println("JoinResultCalculationThread当前线程："+Thread.currentThread().getName()+"开始执行");
        try {
            connection = nodeService.getConnectionByDataSourceName(multiLevelJoinConfig.getNodes().get(currentLevel).getDataSourceName());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("获取数据库连接失败",e);
            System.out.println("JoinResultCalculationThread当前线程："+Thread.currentThread().getName()+"退出");
            return;
        }
        //从页面列表取出一页
        while(true){
            Long offset = currentPageList.remove();
            if(offset==null){
                break;
            }
            //页面数据和父数据集做Join运算
            List<Map<String, Object>>  joinResult = null;
            try {
                joinResult = joinDataSet(offset,parentNodeDatas);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("joinDataSet 调用失败！",e);
            }
            queue.add(joinResult);
        }
        if(connection!=null){
            nodeService.closeConnection(connection);
        }
        System.out.println("JoinResultCalculationThread当前线程："+Thread.currentThread().getName()+"退出");
    }

    /**
     * 结果连接
     * @param offset
     * @param parentNodeDatas
     * @return
     */
    private List<Map<String, Object>> joinDataSet(Long offset, List<Map<String, Object>> parentNodeDatas) throws Exception {

        NodeConfig nodeConfig = multiLevelJoinConfig.getNodes().get(currentLevel);
        //根据偏移量读取表数据一页
        List<Map<String, Object>> tableData = nodeService.getTableData(nodeConfig,offset,connection);


        //获取关联的别名数据集合、
        List<String> joinDataSetNames = getJoinDataSetNames(nodeConfig);
        //校验集合数量，必须等于2
        if(joinDataSetNames.size()!=2)
            throw new Exception("关联集合必须是两个！");
        //取出数量较大的数据集合，构造HashMap
        List<Map<String, Object>> bigDataSet ;
        List<Map<String, Object>> smallDataSet ;
        String bigDataSetName ;
        String smallDataSetName ;

        if(tableData.size()>parentNodeDatas.size()){
            bigDataSet = tableData;
            bigDataSetName = nodeConfig.getAlias();
            smallDataSetName = parentNodeDataAlias;
            smallDataSet =  parentNodeDatas;
        }else{
            smallDataSet = tableData;
            smallDataSetName = nodeConfig.getAlias();
            bigDataSetName  = parentNodeDataAlias;
            bigDataSet =  parentNodeDatas;
        }
        //根据关联字段转换为Map
        Map<String, List<Map<String, Object>>> map = nodeService.getMap(nodeConfig.getJoinColumns(), bigDataSet,bigDataSetName);
        List<Map<String, Object>> mergeDataSet = new ArrayList<>();
        //循环遍历数据集合数量量较小，寻找匹配的记录，收集放到join 关联后的别名数据集合
        // 遍历小列表，查找匹配项
        List<String> fieldNames = nodeService.getJoinColumnNames(nodeConfig.getJoinColumns(), smallDataSetName);
        Map<String,Object> testMap = JSON.parseObject(nodeConfig.getJoinResultFields(), HashMap.class);
        smallDataSet.forEach(smallMap -> {
            String key = nodeService.makeKey(smallMap, fieldNames);
            if(traceLostData(smallMap)){
                System.out.println("丢失数据："+smallMap);
            }
            List<Map<String, Object>> bigMaps = map.get(key);
            if(bigMaps!=null) {
                for (Map<String, Object> bigMap : bigMaps) {
                    if (bigMap != null) {
                        //执行收集
                        try {
                            if(traceLostData(bigMap)){
                                System.out.println("丢失数据："+bigMap);
                            }
                            Map<String, Object> mergeMap = nodeService.mergeMap(bigMap, smallMap, testMap, bigDataSetName, smallDataSetName);
                            mergeDataSet.add(mergeMap);
                        } catch (Exception e) {
                            e.printStackTrace();
                            log.error("mergeMap error:{}", e.getMessage());
                        }
                    }
                }
            }
        });
        return mergeDataSet;
    }
    private boolean traceLostData(Map<String, Object> currentNodeData) {
        if(currentNodeData.get("schemeId")==null){
            return false;
        }
        String schemeId = currentNodeData.get("schemeId").toString();
        if(schemeId.equals("2479355402210705408")||
                schemeId.equals("7244032309306888192")||
                schemeId.equals("7307746228067995648")){
            return true;
        }

        return false;
    }
    /**
     * 获取关联的别名数据集合
     * @param nodeConfig
     * @return
     */
    private List<String> getJoinDataSetNames(NodeConfig nodeConfig) {
        String joinParis[] = nodeConfig.getJoinColumns().split(",");
        List<String> joinDataSetNames = new ArrayList<>();
        String joinDataSetName;
        for(String joinPair:joinParis){
            String[] joinDataSetName1 = joinPair.split("=");
            for(String joinDataSetName2:joinDataSetName1) {
                int startPos = joinDataSetName2.indexOf(".");
                if (startPos >= 0) {
                    joinDataSetName = joinDataSetName2.substring(0, startPos);
                    if (joinDataSetName != null) {
                        joinDataSetName = joinDataSetName.toLowerCase().trim();
                        if (!joinDataSetNames.contains(joinDataSetName)) {
                            joinDataSetNames.add(joinDataSetName);
                        }
                    }
                }
            }
        }
        return joinDataSetNames;
    }

}
